package com.github.binarylei.curator.barrier;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.atomic.AtomicValue;
import org.apache.curator.framework.recipes.atomic.DistributedAtomicInteger;
import org.apache.curator.framework.recipes.barriers.DistributedDoubleBarrier;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.retry.RetryNTimes;
import org.junit.Test;

import java.util.Random;

/**
 * 分布式计数器
 * @author: leigang
 * @version: 2018-04-06
 */
public class ZkCuratorBarrier1 {

    private static final String CONNECT_ADDR = "127.0.0.1:2181";
    private static final int SEESION_OUTTIME = 5 * 1000;

    public static CuratorFramework createCuratorFramework() {
        //1. 重试策略：初试时间为1s，重试10次
        ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 10);

        //2. 通过工厂创建连接
        CuratorFramework cf = CuratorFrameworkFactory.builder()
                .connectString(CONNECT_ADDR)
                .sessionTimeoutMs(SEESION_OUTTIME)
                .retryPolicy(retryPolicy)
                .build();

        return cf;
    }

    @Test
    public void test() throws InterruptedException {
        for (int i = 0; i < 5; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        CuratorFramework cf = createCuratorFramework();
                        cf.start();

                        DistributedDoubleBarrier barrier = new DistributedDoubleBarrier(cf, "/barrier", 5);
                        Thread.sleep(1000 * (new Random().nextInt(3)));
                        System.out.println(Thread.currentThread().getName() + "已经准备");

                        barrier.enter();
                        System.out.println("同时开始运行...");
                        Thread.sleep(1000 * (new Random().nextInt(3)));
                        System.out.println(Thread.currentThread().getName() + "运行完毕");
                        barrier.leave();
                        System.out.println("同时退出运行...");
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }, "t" + i).start();

        }

        Thread.sleep(100 * 1000);
    }
}
